package com.kaimary.smartinteract;

import com.kaimary.smartinteract.datadistribution.repository.DataDistributionRepository;
import com.kaimary.smartinteract.history.model.History;
import com.kaimary.smartinteract.history.repository.HistoryRepository;
import com.kaimary.smartinteract.hive.HiveConfig;
import com.kaimary.smartinteract.sourceindex.model.SourceIndex;
import com.kaimary.smartinteract.sourceindex.repository.SourceIndexRepository;
import com.kaimary.smartinteract.user.model.User;
import com.kaimary.smartinteract.user.repository.UserRepository;
import com.kaimary.smartinteract.wordcount.WordCount;
import iforest.IForest;
import net.sf.json.JSONArray;
import net.sf.json.JSONObject;
import org.codehaus.jackson.map.ObjectMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.context.request.async.DeferredResult;

import javax.validation.Valid;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.sql.*;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.*;

import static java.time.OffsetDateTime.now;

//@CrossOrigin(origins = "*", allowedHeaders = "*")
@RestController
@RequestMapping("/api")
public class Controller {
    private Connection conn = HiveConfig.getConnnection();
    PreparedStatement ps = null;

//    SparkConf sparkConf = new SparkConf().setAppName("SparkHive").setMaster("local[2]");
//    JavaSparkContext sc = new JavaSparkContext(sparkConf);
//    HiveContext hiveContext = new HiveContext(sc);

    @Autowired
    UserRepository userRepository;
    @Autowired
    SourceIndexRepository sourceIndexRepository;
    @Autowired
    DataDistributionRepository dataDistributionRepository;
    @Autowired
    HistoryRepository historyRepository;

    /**
     * 历史操作记录的存储与读取
     */
    @CrossOrigin(origins = "*")
    @PostMapping("/hive/savehistory")
    public ResponseEntity<Integer> SaveHistory(@RequestBody JSONObject filters) {
        JSONObject obj = JSONObject.fromObject(filters);
        String tablename = obj.getString("tablename");
        Integer userid = obj.getInt("userid");
        Integer step = obj.getInt("step");
        JSONArray jorecord = obj.getJSONArray("record");
        String record = jorecord.toString().replace("\"", "\\\"");
        Integer previous = obj.getInt("previous");
        Integer dataset_id = null;
//        String[] sWhere = new String[arr.size()];
//        for (int i=0; i<arr.size(); ++i) {
//            JSONObject filter = arr.getJSONObject(i);
//            String column = filter.getString("columnname");
//            if (filter.getInt("type") == Types.BIGINT || filter.getInt("type") == Types.INTEGER ||
//                    filter.getInt("type") == Types.DECIMAL || filter.getInt("type") == Types.DOUBLE) {
//                sWhere[i] = filter.getString("columnname") + ">" + filter.getJSONArray("data").get(0) +
//                        " AND " + filter.getString("columnname") + "<" + filter.getJSONArray("data").get(1);
//            } else {
//                JSONArray items = filter.getJSONArray("data");
//                sWhere[i] = "";
//                if (!items.isEmpty() && filter.getInt("type") == Types.TIMESTAMP) {
//                    sWhere[i] = filter.getString("columnname") +
//                            " BETWEEN '" + filter.getJSONArray("data").get(0) +
//                            "' AND '" + filter.getJSONArray("data").get(1) + "'";
//                } else if (filter.getInt("type") == Types.VARCHAR) {
//                    if (column.contains("itemdesc")) {
//                        System.out.printf("GetAllDataDistributions =========== itemdesc size:%d%n", items.size());
//                        for (int j = 0; j < items.size(); j++) {
//                            sWhere[i] += column + " NOT LIKE " + "'%" + items.get(j) + "%'";
//                            if (j != items.size() - 1) sWhere[i] += " OR ";
//                        }
//                    } else {
//                        for (int j = 0; j < items.size(); j++) {
//                            sWhere[i] += column + "='" + items.get(j) + "'";
//                            if (j != items.size() - 1) sWhere[i] += " OR ";
//                        }
//                    }
//                }
//            }
//            String column_range = sWhere[i];
//            Integer dataset_id=0;
//            String operation = "XXX";
//            String recommended_method = "XXX";
//            SimpleDateFormat formatter = new SimpleDateFormat("yyyy年MM月dd日 HH:mm:ss");
//            Date curDate = new Date(System.currentTimeMillis());//获取当前时间
//            String create_time = formatter.format(curDate);

        Connection conn = null;
        Statement stmt = null;
        String url = "jdbc:mysql://localhost:3306/userdb?characterEncoding=utf8&useSSL=true";
        String user = "root";
        String pass = "root";
        String sql = "";
        try {
            Class.forName("com.mysql.jdbc.Driver");
//                System.out.println();
            conn = DriverManager.getConnection(url, user, pass);
            stmt = conn.createStatement();
            sql = "SELECT dataset_id FROM sourceindex WHERE tablename=\"" + tablename + "\"";
//                sql = "SELECT dataset_id from sourceindex where tablename="+tablename;
            ResultSet rs = stmt.executeQuery(sql);
            rs.next();
            dataset_id = rs.getInt("dataset_id");
//                SimpleDateFormat formatter = new SimpleDateFormat("yyyy年MM月dd日 HH:mm:ss");
            Date curDate = new Date(System.currentTimeMillis());//获取当前时间
//                String create_time = formatter.format(curDate);

            sql = "INSERT INTO `history`(`user_id`, `dataset_id`, `step`, `record`, `previous`)" +
                    "VALUES(" + userid + "," + dataset_id + "," + step + ",\"" + record + "\"," + previous + ")";
            System.out.println("sql=" + sql);
            stmt.execute(sql);
            //获取当前操作节点的id，返回给前端
            sql = "SELECT log_id FROM history WHERE log_id=(SELECT MAX(log_id) FROM history)";
            rs = stmt.executeQuery(sql);
            rs.next();
            Integer log_id = rs.getInt("log_id");
            return new ResponseEntity<Integer>(log_id, HttpStatus.OK);
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        } catch (SQLException e) {
            e.printStackTrace();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    @CrossOrigin(origins = "*")
    @PostMapping("/hive/getrecenthistory")
    public ResponseEntity<?> GetRecentHistory(@RequestBody JSONObject jsonObject) {
        Integer recent_history_num = jsonObject.getInt("recent");
        String tablename1 = jsonObject.getString("tablename");
        Connection conn = null;
        Statement stmt = null;
        String url = "jdbc:mysql://10.141.223.30:3306/userdb?characterEncoding=utf8&useSSL=true";
//        String url = "jdbc:mysql://localhost:3306/userdb?characterEncoding=utf8&useSSL=true";
        String user = "root";
        String pass = "root";
        String sql = "";
        JSONArray historyjson = new JSONArray();
        try {
            Class.forName("com.mysql.jdbc.Driver");
//            System.out.println();
            conn = DriverManager.getConnection(url, user, pass);
            stmt = conn.createStatement();
            sql = "SELECT dataset_id From sourceindex WHERE tablename=\"" + tablename1 + "\"";
            System.out.println("/hive/getrecenthistory ========== sql=" + sql);
            ResultSet rs1 = stmt.executeQuery(sql);
            Integer datasetid = -1;

            if (rs1.wasNull()) {
                System.out.println("/hive/getrecenthistory ============ history not exist!");
            } else {
//                rs1.next();
                if (rs1.next()) {
                    datasetid = rs1.getInt(1);
                } else {
                    System.out.println("/hive/getrecenthistory ============ history not exist!");
                    return new ResponseEntity<JSONArray>(historyjson, HttpStatus.OK);
                }
            }

            for (Integer i = 1; i <= recent_history_num; ++i) {
                sql = "SELECT * \n" +
                        "FROM (\n" +
                        "        SELECT h1.*, (\n" +
                        "                SELECT\n" +
                        "                    count(*)\n" +
                        "                FROM\n" +
                        "                    history h2\n" +
                        "                WHERE\n" +
                        "                    h1.log_id <= h2.log_id\n" +
                        "                    AND h1.dataset_id=" + datasetid.toString() + "\n" +
                        "                    AND h2.dataset_id=" + datasetid.toString() + "\n" +
                        "            ) AS rownum\n" +
                        "        FROM\n" +
                        "            history h1\n" +
                        "    ) h3\n" +
                        "WHERE\n" +
                        "    rownum = " + i.toString();
                ResultSet rs = stmt.executeQuery(sql);
//                System.out.println("num = "+recent_history_num.toString());
//                Integer sum = 0;
                rs.next();
//                ++sum;
//                System.out.println(rs.getObject(8));
                Integer log_id = rs.getInt(1);
                Integer user_id = rs.getInt(2);
                Integer dataset_id = rs.getInt(3);
                Integer step = rs.getInt(4);
                String str_record = rs.getString(5);
                JSONArray record = JSONArray.fromObject(str_record);
                Integer previous = rs.getInt(6);
                Timestamp time = rs.getTimestamp(7);
//                System.out.println(log_id.toString());
//                System.out.println(user_id.toString());
//                System.out.println(dataset_id.toString());
//                System.out.println(step.toString());
//                System.out.println(str_record+' '+record.toString()+' '+previous.toString());

                sql = "SELECT tablename FROM sourceindex WHERE dataset_id=" + dataset_id;
                rs = stmt.executeQuery(sql);
                rs.next();
                String tablename = rs.getString("tablename");
                JSONObject jobj = new JSONObject();
                jobj.put("log_id", log_id);
                jobj.put("user_id", user_id);
                jobj.put("tablename", tablename);
                jobj.put("step", step);
                jobj.put("record", record);
                jobj.put("previous", previous);
                jobj.put("time", time);
                historyjson.add(jobj);
//                System.out.println(jobj.toString());
            }
            System.out.println("GetRecentHistory =========== " + historyjson.toString());

//            System.out.println("step = "+sum.toString());
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        } catch (SQLException e) {
            e.printStackTrace();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return new ResponseEntity<JSONArray>(historyjson, HttpStatus.OK);
    }

    @CrossOrigin(origins = "*")
    @PostMapping("/data/getpreviousid")
    public ResponseEntity<?> GetPreviousID(@RequestBody JSONObject jsid) {
        Integer id = jsid.getInt("id");
        Integer previousid = -1;
        Connection conn = null;
        Statement stmt = null;
        String url = "jdbc:mysql://localhost:3306/userdb?characterEncoding=utf8&useSSL=true";
        String user = "root";
        String pass = "root";
        String sql = "";
        try {
            Class.forName("com.mysql.jdbc.Driver");
//            System.out.println();
            conn = DriverManager.getConnection(url, user, pass);
            stmt = conn.createStatement();
            sql = "SELECT previous FROM history WHERE log_id = " + id.toString();
            ResultSet rs = stmt.executeQuery(sql);
            rs.next();
            previousid = rs.getInt("previous");
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        } catch (SQLException e) {
            e.printStackTrace();
        } catch (Exception e) {
            e.printStackTrace();
        }

        return new ResponseEntity<Integer>(previousid, HttpStatus.OK);
    }

    @CrossOrigin(origins = "*")
    @PostMapping("/data/gethistorybyid")
    public ResponseEntity<?> GetHistoryByID(@RequestBody JSONObject jsid) {
        Integer id = jsid.getInt("id");

        Integer previousid = -1;
        Connection conn = null;
        Statement stmt = null;
        String url = "jdbc:mysql://localhost:3306/userdb?characterEncoding=utf8&useSSL=true";
        String user = "root";
        String pass = "root";
        String sql = "";
        JSONObject jobj = new JSONObject();
        try {
            Class.forName("com.mysql.jdbc.Driver");
//            System.out.println();
            conn = DriverManager.getConnection(url, user, pass);
            stmt = conn.createStatement();
            sql = "SELECT * FROM history WHERE log_id = " + id.toString();
            ResultSet rs = stmt.executeQuery(sql);
            rs.next();
            Integer log_id = rs.getInt(1);
            Integer user_id = rs.getInt(2);
            Integer dataset_id = rs.getInt(3);
            Integer step = rs.getInt(4);
            String str_record = rs.getString(5);
            JSONArray record = JSONArray.fromObject(str_record);
            Integer previous = rs.getInt(6);
            Timestamp time = rs.getTimestamp(7);
//                System.out.println(log_id.toString());
//                System.out.println(user_id.toString());
//                System.out.println(dataset_id.toString());
//                System.out.println(step.toString());
//                System.out.println(str_record+' '+record.toString()+' '+previous.toString());

            sql = "SELECT tablename FROM sourceindex WHERE dataset_id=" + dataset_id;
            rs = stmt.executeQuery(sql);
            rs.next();
            String tablename = rs.getString("tablename");

            jobj.put("log_id", log_id);
            jobj.put("user_id", user_id);
            jobj.put("tablename", tablename);
            jobj.put("step", step);
            jobj.put("record", record);
            jobj.put("previous", previous);
            jobj.put("time", time);
//            historyjson.add(jobj);
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        } catch (SQLException e) {
            e.printStackTrace();
        }

        return new ResponseEntity<JSONObject>(jobj, HttpStatus.OK);
    }

    private JSONArray GetAllDataDistributions(JSONArray filters) {
        JSONArray arr = JSONArray.fromObject(filters);

        JSONArray distArr = new JSONArray();
//        Integer type=0;
        String columnname = "";
        if (!arr.isEmpty()) {
            long start = System.currentTimeMillis();

//            String sql = "SELECT * FROM " + arr.getJSONObject(0).getString("tablename") + " TABLESAMPLE(BUCKET 1 OUT OF 100 ON rand())";
            String sql = "SELECT * FROM " + arr.getJSONObject(0).getString("tablename");
            String[] sWhere = new String[arr.size()];
            for (int i = 0; i < arr.size(); i++) {
                JSONObject filter = arr.getJSONObject(i);
                System.out.printf("GetAllDataDistributions =========== filter%d:%s%n", i, filter);
//                type = filter.getInt("type");
                columnname = filter.getString("columnname");
                if (filter.getInt("type") == Types.BIGINT || filter.getInt("type") == Types.INTEGER ||
                        filter.getInt("type") == Types.DECIMAL || filter.getInt("type") == Types.DOUBLE) {
                    sWhere[i] = filter.getString("columnname") + ">" + filter.getJSONArray("data").get(0) +
                            " AND " + filter.getString("columnname") + "<" + filter.getJSONArray("data").get(1);
                } else {
                    JSONArray items = filter.getJSONArray("data");
                    sWhere[i] = "";
                    if (!items.isEmpty() && filter.getInt("type") == Types.TIMESTAMP) {
                        System.out.println(items.size() - 2);
                        sWhere[i] = filter.getString("columnname") +
                                " BETWEEN '" + filter.getJSONArray("data").get(0) +
                                "' AND '" + filter.getJSONArray("data").get(1) + "'";
                    } else if (filter.getInt("type") == Types.VARCHAR) {
                        if (columnname.contains("itemdesc")) {
                            System.out.printf("GetAllDataDistributions =========== itemdesc size:%d%n", items.size());
                            for (int j = 0; j < items.size(); j++) {
                                sWhere[i] += columnname + " NOT LIKE " + "'%" + items.get(j) + "%'";
                                if (j != items.size() - 1) sWhere[i] += " OR ";
                            }
                        } else {
                            for (int j = 0; j < items.size(); j++) {
                                sWhere[i] += columnname + "='" + items.get(j) + "'";
                                if (j != items.size() - 1) sWhere[i] += " OR ";
                            }
                        }
                    }
                }
                System.out.printf("GetAllDataDistributions =========== WHERE clause%d=%s%n", i, sWhere[i]);
            }
            sql += " WHERE ";
            boolean first_clause = true;
            for (int i = 0; i < filters.size(); i++) {
                if (!sWhere[i].isEmpty()) {
                    if (first_clause) {
                        first_clause = false;
                    } else {
                        sql += " AND ";
                    }
                    sql += "(" + sWhere[i] + ")";
                }
            }
            System.out.printf("GetAllDataDistributions =========== sql=%s%n", sql);
            System.out.println("GetAllDataDistributions =========== SQL generation:" + (System.currentTimeMillis() - start) + "ms");
            start = System.currentTimeMillis();
            try {
                PreparedStatement ps = HiveConfig.prepare(conn, sql);
                ResultSet rs = ps.executeQuery();

//                System.out.println(rs.wasNull());
//                System.out.println(rs.next());
                if (!rs.next()) {
                    for (int i = 0; i < arr.size(); ++i) {
                        JSONObject obj = new JSONObject();
                        obj.put("columnname", arr.getJSONObject(i).get("columnname"));
                        obj.put("type", arr.getJSONObject(i).get("type"));
                        obj.put("data", new JSONArray());
                        distArr.add(obj);
                    }
//                    distArr = new JSONArray();
                    System.out.println("GetAllDataDistributions =========== EMPTY RESULT!");
                } else {
//                    System.out.println(rs.getString(1));
                    int num = rs.getMetaData().getColumnCount();

                    ArrayList<?>[] data = new ArrayList[num];
                    // Retrieve all the values into array by column.
                    boolean first = true;
                    for (int index = 0; index < num; index++) {
                        //                    Thread t = new Thread(() ->{});
                        Integer type = rs.getMetaData().getColumnType(index + 1);
                        if (type == Types.BIGINT || type == Types.INTEGER) {
                            if (first) data[index] = new ArrayList<Integer>();
                            ((ArrayList<Integer>) data[index]).add(rs.getInt(index + 1));
                        } else if (type == Types.DOUBLE || type == Types.DECIMAL) {
//                            System.out.println(rs.getDouble(index+1));
                            if (first) data[index] = new ArrayList<Double>();
                            ((ArrayList<Double>) data[index]).add(rs.getDouble(index + 1));
                        } else if (type == Types.TIMESTAMP) {
                            if (first) data[index] = new ArrayList<Timestamp>();
                            ((ArrayList<Timestamp>) data[index]).add(rs.getTimestamp(index + 1));
                        } else {
                            if (first) data[index] = new ArrayList<String>();
                            ((ArrayList<String>) data[index]).add(rs.getString(index + 1));
                        }
                    }
                    first = false;
                    while (rs.next()) {
                        for (int index = 0; index < num; index++) {
                            //                    Thread t = new Thread(() ->{});
                            Integer type = rs.getMetaData().getColumnType(index + 1);
                            if (type == Types.BIGINT || type == Types.INTEGER) {
                                if (first) data[index] = new ArrayList<Integer>();
                                ((ArrayList<Integer>) data[index]).add(rs.getInt(index + 1));
                            } else if (type == Types.DOUBLE || type == Types.DECIMAL) {
//                                System.out.println(rs.getDouble(index + 1));
                                if (first) data[index] = new ArrayList<Double>();
                                ((ArrayList<Double>) data[index]).add(rs.getDouble(index + 1));
                            } else if (type == Types.TIMESTAMP) {
                                if (first) data[index] = new ArrayList<Timestamp>();
                                ((ArrayList<Timestamp>) data[index]).add(rs.getTimestamp(index + 1));
                            } else {
                                if (first) data[index] = new ArrayList<String>();
                                ((ArrayList<String>) data[index]).add(rs.getString(index + 1));
                            }
                        }
                        first = false;
                    }
                    System.out.println("GetAllDataDistributions =========== Retrieve all data:" + (System.currentTimeMillis() - start) + "ms");
                    // Calculate the xAxis and yAxis distributions for each column
                    ArrayList<String>[] xAxis = new ArrayList[num];
                    ArrayList<Integer>[] yAxis = new ArrayList[num];
                    for (int index = 0; index < num; index++) {
                        start = System.currentTimeMillis();
                        first = true;
                        int i = 0;
                        int type = rs.getMetaData().getColumnType(index + 1);
                        if (type == Types.BIGINT || type == Types.INTEGER) {
                            ArrayList<Integer> rows = (ArrayList<Integer>) data[index];
                            int min = Collections.min(rows);
                            int max = Collections.max(rows);
                            Collections.sort(rows);

                            xAxis[index] = new ArrayList<String>();
                            yAxis[index] = new ArrayList<Integer>();
                            if (max - min < 50) {
                                for (int j = 0; j < max - min; ++j) {
                                    xAxis[index].add(String.valueOf(min + j));
                                    int count = 0;
                                    while (i < data[index].size() && rows.get(i) == min + j) {
                                        ++i;
                                        ++count;
                                    }
                                    yAxis[index].add(count);
                                }
                            } else {
                                for (int j = 0; j < 50; ++j) {
                                    xAxis[index].add((String.valueOf(min + j * (max - min) / 50) + "-" + String.valueOf(min + (j + 1) * (max - min) / 50)));
                                    int count = 0;
                                    while (i < data[index].size() && rows.get(i) <= min + (j + 1) * (max - min) / 50) {
                                        ++i;
                                        ++count;
                                    }
                                    yAxis[index].add(count);
                                }
                            }
                        } else if (type == Types.DECIMAL || type == Types.DOUBLE) {
                            ArrayList<Double> rows = (ArrayList<Double>) data[index];
//                            System.out.println(rows.get(1));
                            Double min = Collections.min(rows);
                            Double max = Collections.max(rows);
                            Collections.sort(rows);
                            System.out.printf("GetAllDataDistributions ===========%s%n", min);
                            System.out.printf("GetAllDataDistributions ===========%s%n", max);
                            xAxis[index] = new ArrayList<String>();
                            yAxis[index] = new ArrayList<Integer>();
                            if (max - min < 50) {
                                for (int j = 0; j <= (int) (max - min); ++j) {
                                    xAxis[index].add(String.valueOf(min + j));
                                    int count = 0;
                                    while (i < data[index].size() && rows.get(i) <= min + j) {
                                        ++i;
                                        ++count;
                                    }
                                    yAxis[index].add(count);
                                }
                            } else {
                                for (int j = 0; j < 50; ++j) {
                                    xAxis[index].add((String.valueOf(min + j * (max - min) / 50) + "-" + String.valueOf(min + (j + 1) * (max - min) / 50)));
                                    int count = 0;
                                    while (i < data[index].size() && rows.get(i) <= min + (j + 1) * (max - min) / 50) {
                                        ++i;
                                        ++count;
                                    }
                                    yAxis[index].add(count);
                                }
                            }
                        } else if (type == Types.TIMESTAMP) {
                            ArrayList<Date> rows = (ArrayList<Date>) data[index];
                            Date min = Collections.min(rows);
                            Date max = Collections.max(rows);
                            //                        System.out.println("Min = "+min);
                            //                        System.out.println("Max = "+max);
                            Collections.sort(rows);
                            xAxis[index] = new ArrayList<String>();
                            yAxis[index] = new ArrayList<Integer>();

                            long gap = max.getTime() - min.getTime();
                            System.out.printf("GetAllDataDistributions =========== gap: %d%n", gap);
                            Integer days = (int) (gap / (24 * 60 * 60 * 1000));
                            System.out.printf("GetAllDataDistributions =========== days interval:%d%n", days);
                            if (days < 50) {
                                for (int j = 0; j < days + 1; ++j) {
                                    Calendar c = Calendar.getInstance();
                                    c.setTime(min);
                                    c.add(Calendar.DAY_OF_MONTH, j);
                                    Date d = c.getTime();
                                    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.S");

                                    String sf = sdf.format(d);
                                    //                                System.out.println(sf+","+rows.get(0));
                                    xAxis[index].add(sf);
                                    d = sdf.parse(sf);
                                    int count = 0;
                                    while (i < data[index].size() && (rows.get(i)).toString().equals(sf)) {
                                        ++i;
                                        ++count;
                                    }
                                    yAxis[index].add(count);
                                }
                            } else {
                                for (int j = 0; j < 50; ++j) {
                                    Calendar c = Calendar.getInstance();
                                    Calendar c1 = Calendar.getInstance();
                                    c.setTime(min);
                                    c1.setTime(min);
                                    c.add(Calendar.DAY_OF_MONTH, j * days / 50);
                                    c1.add(Calendar.DAY_OF_MONTH, (j + 1) * days / 50);
                                    Date d = c.getTime();
                                    Date d1 = c1.getTime();
                                    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
                                    String sf = sdf.format(d);
                                    String sf1 = sdf.format(d1);

                                    xAxis[index].add(sf + "-" + sf1);
                                    //                                System.out.printf("GetAllDataDistributions =========== %s-%s%n", sf,sf1);
                                    d1 = sdf.parse(sf1);
                                    int count = 0;
                                    while (i < data[index].size() && rows.get(i).before(d1)) {
                                        ++i;
                                        ++count;
                                    }
                                    yAxis[index].add(count);
                                }
                            }
                        } else {
                            ArrayList<String> rows = (ArrayList<String>) data[index];
                            String column = rs.getMetaData().getColumnName(index + 1).split("\\.")[1];
                            if (column.equals("itemdesc")) {
                                System.out.println("GetAllDataDistributions =========== word count processing...");
                                String text = "";
                                while (i < data[index].size()) {
                                    text += " ";
                                    text += rows.get(i++);
                                }
                                CharTreeNode root = geneCharTree(text);
                                List result = new ArrayList();
                                getWordCountFromCharTree(result, root);
                                Collections.sort(result, new Comparator() {
                                    public int compare(Object o1, Object o2) {
                                        WordCount wc1 = (WordCount) o1;
                                        WordCount wc2 = (WordCount) o2;
                                        return wc2.getCount() - wc1.getCount();
                                    }
                                });
                                System.out.printf("GetAllDataDistributions =========== size:%s%n", result.size());
                                xAxis[index] = new ArrayList<String>();
                                yAxis[index] = new ArrayList<Integer>();
                                if (result.size() > 30) {
                                    for (int ii = 10; ii < 30; ii++) {
                                        WordCount wc = (WordCount) result.get(ii);
                                        System.out.printf("GetAllDataDistributions =========== wc:%s %d%n", wc.getWord(), wc.getCount());
                                        xAxis[index].add(wc.getWord());
                                        yAxis[index].add(wc.getCount());
                                    }
                                }
                            } else {
                                Map<String, Integer> map = new HashMap<>();
                                while (i < data[index].size()) {
                                    String item = rows.get(i++);
                                    if (map.containsKey(item)) {
                                        map.put(item, map.get(item) + 1);
                                    } else {
                                        map.put(item, 1);
                                    }
                                }
                                List<Map.Entry<String, Integer>> list = new ArrayList<Map.Entry<String, Integer>>(map.entrySet());
                                //                                if (!columnname.contains("solddate")) {
                                //                                    System.out.println(columnname + "no");
                                //                            List<Map.Entry<String, Integer>> list = new ArrayList<Map.Entry<String, Integer>>(map.entrySet());
                                //非日期数据按条目数排序
                                Collections.sort(list, new Comparator<Map.Entry<String, Integer>>() {
                                    @Override
                                    public int compare(Map.Entry<String, Integer> o1, Map.Entry<String, Integer> o2) {
                                        return o2.getValue().compareTo(o1.getValue());
                                    }
                                });
                                Integer count = 0;
                                Integer sum = 0;
                                xAxis[index] = new ArrayList<String>();
                                yAxis[index] = new ArrayList<Integer>();
                                for (Map.Entry<String, Integer> entry : list) {
                                    ++count;
                                    if (count < 9) {
                                        xAxis[index].add(entry.getKey());
                                        yAxis[index].add(entry.getValue());
                                    } else {
                                        sum += entry.getValue();
                                    }
                                }

                                if (count >= 10 && xAxis[index].size() <= 9) {
                                    xAxis[index].add("Others");
                                    yAxis[index].add(sum);
                                }
                                //                                } else {
                                //                                    System.out.println(columnname + " has date");
                                //                                    //日期数据按日期合并排序
                                //                                    Collections.sort(list, new Comparator<Map.Entry<String, Integer>>() {
                                //                                        @Override
                                //                                        public int compare(Map.Entry<String, Integer> o1, Map.Entry<String, Integer> o2) {
                                //                                            return o1.getKey().compareTo(o2.getKey());
                                //                                        }
                                //                                    });
                                //
                                //                                    Integer count = 0;
                                //                                    Integer sum = 0;
                                //                                    xAxis[index] = new ArrayList<String>();
                                //                                    yAxis[index] = new ArrayList<Integer>();
                                //                                    String left = "";
                                //                                    String right = "";
                                //                                    for (Map.Entry<String, Integer> entry : list) {
                                //                                        if (count == 0) left = entry.getKey();
                                //                                        ++count;
                                //                                        sum += entry.getValue();
                                //                                        if (count >= (list.size() / 10)) {
                                //                                            right = entry.getKey();
                                //                                            xAxis[index].add(left + "~\n" + right);
                                //                                            yAxis[index].add(entry.getValue());
                                //                                            count = 0;
                                //                                            sum = 0;
                                //                                        }
                                //                                    }
                                //                                }
                            }
                        }

                        Iterator<String> itrx = xAxis[index].listIterator();
                        Iterator<Integer> itry = yAxis[index].listIterator();
                        JSONObject dist = new JSONObject();
                        String columanme = rs.getMetaData().getColumnName(index + 1).split("\\.")[1];
                        dist.put("columnname", columanme);
                        dist.put("type", rs.getMetaData().getColumnType(index + 1));
                        Integer in = 0;
                        JSONArray jsarrary = new JSONArray();
                        while (itrx.hasNext()) {
                            ++in;
                            JSONObject jobj = new JSONObject();
                            jobj.put(itrx.next(), itry.next());
                            jsarrary.add(jobj);
                        }
                        dist.put("data", jsarrary);

                        distArr.add(dist);
                        System.out.println("GetAllDataDistributions =========== Format the JSON object for column " + columanme + ":" + (System.currentTimeMillis() - start) + "ms");
                    }
                }
            } catch (SQLException sqle) {
                sqle.printStackTrace();
            } catch (ParseException e) {
                e.printStackTrace();
            }
        }
        return distArr;
    }

    private CharTreeNode geneCharTree(String text) {
        CharTreeNode root = new CharTreeNode();
        CharTreeNode p = root;
        char c = ' ';
        for (int i = 0; i < text.length(); ++i) {
            c = text.charAt(i);
            if (c >= 'A' && c <= 'Z')
                c = (char) (c + 'a' - 'A');
            if (c >= 'a' && c <= 'z') {
                if (p.children[c - 'a'] == null)
                    p.children[c - 'a'] = new CharTreeNode();
                p = p.children[c - 'a'];
            } else {
                p.cnt++;
                p = root;
            }
        }
        if (c >= 'a' && c <= 'z')
            p.cnt++;
        return root;
    }

    /**
     * 使用深度优先搜索遍历单词树并将对应单词放入结果集中
     */
    private void getWordCountFromCharTree(List result, CharTreeNode p, char[] buffer, int length) {
        for (int i = 0; i < 26; ++i) {
            if (p.children[i] != null) {
                buffer[length] = (char) (i + 'a');
                if (p.children[i].cnt > 0) {
                    WordCount wc = new WordCount();
                    wc.setCount(p.children[i].cnt);
                    wc.setWord(String.valueOf(buffer, 0, length + 1));
                    result.add(wc);
                }
                getWordCountFromCharTree(result, p.children[i], buffer, length + 1);
            }
        }
    }

    private void getWordCountFromCharTree(List result, CharTreeNode p) {
        getWordCountFromCharTree(result, p, new char[10000], 0);
    }

    /**
     * 单词树结点的定义
     */
    class CharTreeNode {
        int cnt = 0;
        CharTreeNode[] children = new CharTreeNode[26];
    }

    @CrossOrigin(origins = "*")
    @GetMapping("/user/list")
    public List<User> getAllUsers() {
        return userRepository.findAll();
    }

    @CrossOrigin(origins = "*")
    @PostMapping("/user/add")
    public User createUser(@Valid @RequestBody User user) {
        return userRepository.save(user);
    }

    @CrossOrigin(origins = "*")
    @PostMapping("/user/login")
    public ResponseEntity<?> UserLogin(@Valid @RequestBody User user) {
        System.out.printf("POST /user/login =========== username= %s, password=%s%n", user.getUsername(), user.getPassword());
        User userExists = userRepository.findByUsername(user.getUsername());
        if (userExists != null && (userExists.getPassword()).equals(user.getPassword()) == true) {
            System.out.println("POST /user/login =========== user profile verified!");
            return new ResponseEntity<>(userExists, HttpStatus.OK);
        } else {
            return new ResponseEntity(HttpStatus.NO_CONTENT);
        }
    }

    @CrossOrigin(origins = "*")
    @GetMapping("/index/list")
    public ResponseEntity<?> getAllDataSources() {
        System.out.printf("GET /index/list =========== successfully%n");
        return new ResponseEntity<>(sourceIndexRepository.findAllByOrderByIscommonDescUpdatedAtDesc(), HttpStatus.OK);
    }

    @CrossOrigin(origins = "*")
    @PostMapping("/index/table")
    public ResponseEntity<?> getOneDataSource(@Valid @RequestBody SourceIndex sourceIndex) {
        System.out.printf("POST /index/table =========== tablename=%s%n", sourceIndex.getTablename());
        // Sort the dataset list ordered by "iscommon" property first and then, updated time.
        return new ResponseEntity<>(sourceIndexRepository.findByTablename(sourceIndex.getTablename()), HttpStatus.OK);
    }

    @CrossOrigin(origins = "*")
    @PostMapping("/index/update")
    public ResponseEntity<?> updateIsCommonProperty(@Valid @RequestBody SourceIndex sourceIndex) {
        System.out.printf("POST /index/update =========== tablename=%s%n", sourceIndex.getTablename());
        SourceIndex si = sourceIndexRepository.findByTablename(sourceIndex.getTablename());
        si.setIscommon(sourceIndex.isIscommon());
        sourceIndexRepository.save(si);
        return new ResponseEntity<>(HttpStatus.OK);
    }

    @CrossOrigin(origins = "*")
    @PostMapping("/index/recommend")
    public ResponseEntity<?> DataSourcesRecommend(@RequestBody String jsonData) throws IOException {
        System.out.printf("POST /index/recommend =========== parameter=%s%n", jsonData);
        JSONObject obj = JSONObject.fromObject(jsonData);
        String label = obj.get("label").toString();
        System.out.printf("POST /index/recommend =========== successfully retrieved the user label: %s%n", label);

        JSONObject returnData = new JSONObject();
        returnData.put("status", "success");

        List<SourceIndex> list = sourceIndexRepository.findAllByLabel(label);
        returnData.put("count", list.size());
        System.out.printf("POST /index/recommend =========== recommend %d datasets for this user...%n", list.size());
        ObjectMapper mapper = new ObjectMapper();
        returnData.put("data", mapper.writeValueAsString(list));
        return new ResponseEntity<JSONObject>(returnData, HttpStatus.OK);
    }

    @CrossOrigin(origins = "*")
    @PostMapping("/hive/rowcount")
    public ResponseEntity<?> GetRowCount(@RequestBody String jsonData) throws ClassNotFoundException, SQLException {
        System.out.printf("POST /hive/rowcount =========== parameter=%s%n", jsonData);
        JSONObject obj = JSONObject.fromObject(jsonData);
        String sql = "SELECT COUNT(*) AS rowcount FROM " + obj.getString("tablename");

        System.out.printf("POST /hive/rowcount =========== sql=%s%n", sql);
        try {
            ps = conn.prepareStatement(sql);
            long start = System.currentTimeMillis();
            ResultSet rs = ps.executeQuery();
            System.out.println("POST /hive/rowcount =========== execution time for sql: " + (System.currentTimeMillis() - start) + "ms");

            start = System.currentTimeMillis();
            rs.next();
            int total = rs.getInt("rowcount");
            System.out.println("POST /hive/rowcount =========== execution time for traverse: " + (System.currentTimeMillis() - start) + "ms");

            rs.close();
            JSONObject returnData = new JSONObject();
            returnData.put("total", total);
            return new ResponseEntity<JSONObject>(returnData, HttpStatus.OK);
        } catch (SQLException e) {
            e.printStackTrace();
        }
        return new ResponseEntity<>(HttpStatus.NO_CONTENT);
    }

    @CrossOrigin(origins = "*")
    @PostMapping("/data/getsumbydate")
    public ResponseEntity<?> GetSumByDate(@RequestBody String jsonData) throws SQLException {
        JSONObject jsonObject = JSONObject.fromObject(jsonData);
        String tablename = jsonObject.getString("tablename");
        String date = jsonObject.getString("date");
        String column = jsonObject.getString("column");
        String sumtype = jsonObject.getString("sumtype");
        String sql = "";
        try {
            JSONObject returnData = new JSONObject();
            if (sumtype.equals("interval")) {
                if (column.equals("quantity")) {
                    sql = "SELECT quantity FROM " + tablename + " WHERE solddate BETWEEN '2001-01-01' AND '" + date + "'";

                } else if (column.equals("price")) {
                    sql = "SELECT price,quantity FROM " + tablename + " WHERE solddate BETWEEN '2001-01-01' AND '" + date + "'";

                } else {
                    System.out.println("getsumbydate ========== wrong column name! only quantity or price accepted!");
                    return new ResponseEntity<>(HttpStatus.NO_CONTENT);
                }
            } else if (sumtype.equals("exact")) {
                if (column.equals("quantity")) {
                    sql = "SELECT quantity FROM " + tablename + " WHERE solddate BETWEEN '" + date + "' AND '" + date + "'";

                } else if (column.equals("price")) {
                    sql = "SELECT price,quantity FROM " + tablename + " WHERE solddate BETWEEN '" + date + "' AND '" + date + "'";
                } else {
                    System.out.println("getsumbydate ========== wrong column name! only quantity or price accepted!");
                    return new ResponseEntity<>(HttpStatus.NO_CONTENT);
                }
            } else {
                System.out.println("getsumbydate ========== wrong sum type! only interval or exact accepted!");
                return new ResponseEntity<>(HttpStatus.NO_CONTENT);
            }
            ps = conn.prepareStatement(sql);
            ResultSet rs = ps.executeQuery();
            Double sum = 0.0;
            while (rs.next()) {
                sum += rs.getDouble(1);
            }
            returnData.put("sum", sum);
            System.out.println("getsumbydate ========== sum=" + sum.toString());

            return new ResponseEntity<JSONObject>(returnData, HttpStatus.OK);
        } catch (SQLException e) {
            e.printStackTrace();
        }
        return new ResponseEntity<>(HttpStatus.NO_CONTENT);
    }

    @CrossOrigin(origins = "*")
    @PostMapping("/hive/executesql")
    public ResponseEntity<?> ExecuteSQL(@RequestBody String sql) throws SQLException {
        System.out.printf("POST /hive/executesql =========== sql=%s%n", sql);
        ps = conn.prepareStatement(sql);
        ResultSet rs = ps.executeQuery();
//        String result = hiveContext.sql(sql).collect().toString();
        JSONObject returnData = new JSONObject();
//        returnData.put("data",result);
        try {
            int size = 0;
//          ArrayList<ArrayList<String>> collection = new ArrayList<ArrayList<String>>();
            List<JSONObject> rows = new ArrayList<JSONObject>();
            int columns = rs.getMetaData().getColumnCount();

//            start = System.currentTimeMillis();
            while (rs.next()) {
                size++;
                if (size <= 20) {
                    ArrayList<String> arr = new ArrayList<String>();
                    JSONObject row = new JSONObject();
                    for (int i = 1; i <= columns; i++) {
                        //arr.add(rs.getString(i));
                        row.put(rs.getMetaData().getColumnName(i).split("\\.")[1], rs.getString(i));
                    }
                    rows.add(row);
                }
                //collection.add(arr);
            }
            returnData.put("rowcount", size);
            returnData.put("data", rows);
//            System.out.println("POST /hive/getdata =========== execution time for traverse: " + (System.currentTimeMillis() - start) + "ms");

            return new ResponseEntity<JSONObject>(returnData, HttpStatus.OK);
        } catch (SQLException e) {
            e.printStackTrace();
        }
        return new ResponseEntity<>(HttpStatus.NO_CONTENT);
    }

    @CrossOrigin(origins = "*")
    @PostMapping("/hive/gettableschema")
    public ResponseEntity<?> GetTableSchema(@RequestBody String tablename) throws SQLException {
        System.out.printf("POST /hive/gettableschema =========== tablename=%s%n", tablename);
        try {
            String sql = "DESC " + tablename;
            ps = conn.prepareStatement(sql);
            ResultSet rs = ps.executeQuery();
            JSONObject returnData = new JSONObject();
//            int size = 0;
//          ArrayList<ArrayList<String>> collection = new ArrayList<ArrayList<String>>();
//            List<JSONObject> rows = new ArrayList<JSONObject>();
            int columns = rs.getMetaData().getColumnCount();

//            start = System.currentTimeMillis();
            while (rs.next()) {
//                ArrayList<String> arr = new ArrayList<String>();
//                JSONObject row = new JSONObject();
//                System.out.println(returnData.toString());
//                if (!rs. && rs.getString(1).charAt(0)!='#')
                returnData.put(rs.getString(1), rs.getString(2));
//                rows.add(row);
                //collection.add(arr);
            }
            Iterator<String> keys = returnData.keys();
            Set<String> removekeys = new HashSet<>();
            while (keys.hasNext()) {
                String key = keys.next();
                if (key.charAt(0) == '#') {
                    removekeys.add(key);
                }
            }
            Iterator<String> it = removekeys.iterator();
            while (it.hasNext()) {
                String key = it.next();
                returnData.remove(key);
            }
//            returnData.put("rowcount", size);
//            returnData.put("data", rows);
//            System.out.println("POST /hive/getdata =========== execution time for traverse: " + (System.currentTimeMillis() - start) + "ms");
            System.out.println("POST /hive/gettableschema =========== schema=" + returnData.toString());
            return new ResponseEntity<JSONObject>(returnData, HttpStatus.OK);
        } catch (SQLException e) {
            e.printStackTrace();
        }
        return new ResponseEntity<>(HttpStatus.NO_CONTENT);
    }

    @CrossOrigin(origins = "*")
    @PostMapping("/hive/getdata")
    public ResponseEntity<?> GetTableData(@RequestBody String jsonData) throws ClassNotFoundException, SQLException {
        System.out.printf("POST /hive/getdata =========== parameter=%s%n", jsonData);
        JSONObject obj = JSONObject.fromObject(jsonData);
        int page = obj.getInt("page");
        String sql = "SELECT " + obj.getString("tablename") + ".* FROM " + obj.getString("tablename");

        JSONArray filters = obj.getJSONArray("filters");
        if (!filters.isEmpty()) {
            String[] sWhere = new String[filters.size()];
            for (int i = 0; i < filters.size(); i++) {
                sWhere[i] = "";
                JSONObject obj1 = (JSONObject) filters.get(i);

                if (obj1.getInt("type") == Types.BIGINT || obj1.getInt("type") == Types.INTEGER
                        || obj1.getInt("type") == Types.FLOAT || obj1.getInt("type") == Types.DOUBLE
                ) {
//                    数值型数据区间
                    sWhere[i] += obj1.getString("columnname") + ">=" + obj1.getJSONArray("data").get(0) +
                            " AND " + obj1.getString("columnname") + "<=" + obj1.getJSONArray("data").get(1);
                } else if (obj1.getInt("type") == Types.TIMESTAMP) {
//                    时间区间
                    sWhere[i] += obj1.getString("columnname") + " BETWEEN \'" + obj1.getJSONArray("data").get(0) +
                            "\' AND \'" + obj1.getJSONArray("data").get(1) + "\'";
                } else if (obj1.getInt("type") == Types.VARCHAR) {
//                    字符串型数据
                    JSONArray arr = obj1.getJSONArray("data");
                    //sWhere[i] = "";
//                    sWhere[i] += "(";
                    for (int j = 0; j < arr.size(); j++) {
                        sWhere[i] += obj1.getString("columnname") + " LIKE '%" + arr.get(j) + "%'";
                        if (j != arr.size() - 1) sWhere[i] += " OR ";
                    }
//                    sWhere[i] += ")";
                }
                System.out.printf("POST /hive/getdata =========== WHERE clause%d=%s%n", i, sWhere[i]);
            }

            sql += " WHERE ";
            for (int i = 0; i < filters.size(); i++) {
//                System.out.println(sWhere[i]);
                sql += "(" + sWhere[i] + ")";
                if (i != filters.size() - 1) sql += " AND ";
            }
        }
        sql += " LIMIT " + page * 20;
        System.out.printf("POST /hive/getdata =========== sql=%s, page=%d%n", sql, page);
        try {
            long start = System.currentTimeMillis();
            PreparedStatement ps = HiveConfig.prepare(conn, sql);
            ResultSet rs = ps.executeQuery();
            System.out.println("POST /hive/getdata =========== execution time for sql: " + (System.currentTimeMillis() - start) + "ms");

            JSONObject returnData = new JSONObject();
            // Cusor to the position of current page.
            if (page > 1) {
                int num = (page - 1) * 20;
                while ((num--) > 0) {
                    rs.next();
                }
            }
            int size = 0;
            ArrayList<ArrayList<String>> collection = new ArrayList<ArrayList<String>>();
            List<JSONObject> rows = new ArrayList<JSONObject>();
            int columns = rs.getMetaData().getColumnCount();

            start = System.currentTimeMillis();
            while (rs.next() && size < 20) {
                size++;
                ArrayList<String> arr = new ArrayList<String>();
                JSONObject row = new JSONObject();
                for (int i = 1; i <= columns; i++) {
                    //arr.add(rs.getString(i));
                    row.put(rs.getMetaData().getColumnName(i).split("\\.")[1], rs.getString(i));
                }
                rows.add(row);
                //collection.add(arr);
            }

            returnData.put("data", rows);
            System.out.println("POST /hive/getdata =========== execution time for traverse: " + (System.currentTimeMillis() - start) + "ms");

            return new ResponseEntity<JSONObject>(returnData, HttpStatus.OK);
        } catch (SQLException e) {
            e.printStackTrace();
        }
        return new ResponseEntity<>(HttpStatus.NO_CONTENT);
    }

    @CrossOrigin(origins = "*")
    @PostMapping("/data/onlinedist")
    public ResponseEntity<?> GetOnlineDistribution(@RequestBody String jsonData) {
//        JSONObject obj = JSONObject.fromObject(jsonData);
//        String sql = "SELECT " + obj.getString("columnname") + " FROM " + obj.getString("tablename");
//        System.out.printf("POST data/onlinedist =========== sql=%s%n", sql);
        return new ResponseEntity<JSONArray>(GetAllDataDistributions(JSONArray.fromObject(jsonData)), HttpStatus.OK);
//        JSONArray data = obj.getJSONArray("data");
//        try {
//            PreparedStatement ps = HiveConfig.prepare(conn, sql);
//            ResultSet rs = ps.executeQuery();
//
//            ArrayList<String> xAxis = new ArrayList<String>();
//            ArrayList<Integer> yAxis = new ArrayList<Integer>();
//
//            int type = rs.getMetaData().getColumnType(1);
//            if (type == Types.BIGINT || type == Types.INTEGER) {
//                System.out.printf("POST data/onlinedist =========== type: INTEGER%n");
//                int max = (int) data.get(1);
//                int min = (int) data.get(0);
//
//                ArrayList<Integer> si = new ArrayList<Integer>();
//                while (rs.next()) {
//                    if (type == Types.BIGINT || type == Types.INTEGER) {
//                        si.add(rs.getInt(1));
//                    }
//                }
//
//                if (!si.isEmpty()) {
//                    System.out.printf("POST data/onlinedist =========== min=%d;max=%d%n", min, max);
//
//                    Collections.sort(si);
//                    int index = 0;
//                    while (index < si.size() && si.get(index) < min) ++index;
//                    for (int i = 0; i < 10; ++i) {
//                        xAxis.add((String.valueOf(min + i * (max - min) / 10) + "-" + String.valueOf(min + (i + 1) * (max - min) / 10)));
//                        int count = 0;
//                        while (index < si.size() && si.get(index) <= max && si.get(index) <= min + (i + 1) * (max - min) / 10) {
//                            ++index;
//                            ++count;
//                        }
//                        yAxis.add(count);
//                    }
//                }
//            } else {
//                System.out.printf("POST data/onlinedist =========== type: STRING%n");
//                Map<String, Integer> map = new HashMap<>();
//
//                for (int j = 0; j < data.size(); j++) {
//                    map.put((String) data.get(j), 0);
//                }
//                try {
//                    while (rs.next()) {
//                        String skey = rs.getString(1);
//                        if (map.containsKey(skey)) {
//                            map.put(skey, map.get(skey) + 1);
//                        }
//                    }
//                } catch (SQLException e) {
//                    e.printStackTrace();
//                }
//                for (Map.Entry<String, Integer> entry : map.entrySet()) {
//                    xAxis.add(entry.getKey());
//                    yAxis.add(entry.getValue());
//                }
//            }
//
//            JSONObject charOptions = new JSONObject();
//            Iterator<String> itrx = xAxis.listIterator();
//            Iterator<Integer> itry = yAxis.listIterator();
//            JSONArray jsarray = new JSONArray();
//            Integer index = 0;
//            while (itrx.hasNext()) {
//                ++index;
//                JSONObject obj1 = new JSONObject();
//                obj1.put(itrx.next(), itry.next());
//                jsarray.add(obj1);
//                //charOptions.put(new String("sec" + index.toString()), jobj);
//            }
//            charOptions.put("columnname", obj.getString("columnname"));
//            charOptions.put("data", jsarray);
//            System.out.println(charOptions.toString());
//            return new ResponseEntity<JSONObject>(charOptions, HttpStatus.OK);

//        } catch (SQLException sqle) {
//            sqle.printStackTrace();
//        }
//        return new ResponseEntity<>(HttpStatus.NO_CONTENT);
    }

    //The API is obsoleted. Please use GetDeepeyeRecommednation instead.
    @CrossOrigin(origins = "*")
    @PostMapping("/data/onlinerec")
    public ResponseEntity<?> GetOnlineRecommendation(@RequestBody String jsonData) {
        JSONArray jsonArray = JSONArray.fromObject(jsonData);
        JSONArray distArray = new JSONArray();
        /*
        *********************************************************************
            The algorithm for distribution recommend
        *********************************************************************
        */
        Random r = new Random();
        for (int i = 0; i < 4; i++) {
            int x = r.nextInt(jsonArray.size());
            if (x == 4) x++;
            //Y-Axis only supports for Interger and Double type.
            int y = r.nextInt(2);
            y = 2 + y;
            if (x == y) x = 5;

            System.out.printf("POST /data/onlinerec =========== x=%d,y=%d%n", x, y);
            JSONObject obj_x = JSONObject.fromObject(jsonArray.get(x));
            JSONObject obj_y = JSONObject.fromObject(jsonArray.get(y));
            String sql = "SELECT " + obj_x.getString("columnname") + "," + obj_y.getString("columnname") + " FROM " + obj_x.getString("tablename");
            System.out.printf("POST /data/onlinerec =========== sql=%s%n", sql);

            JSONArray filter_x = obj_x.getJSONArray("data");
            JSONArray filter_y = obj_y.getJSONArray("data");

            ArrayList<String> xAxis = new ArrayList<String>();
            ArrayList<Integer> yAxis = new ArrayList<Integer>();
            try {
                PreparedStatement ps = HiveConfig.prepare(conn, sql);
                ResultSet rs = ps.executeQuery();

                int type = rs.getMetaData().getColumnType(1);
                if (type == Types.BIGINT || type == Types.INTEGER) {
                    int max_x = (int) filter_x.get(1);
                    int min_x = (int) filter_x.get(0);
                    int max_y = (int) filter_y.get(1);
                    int min_y = (int) filter_y.get(0);
                    System.out.printf("POST /data/onlinerec =========== type: INTEGER%n");

                    for (int j = 0; j < 10; ++j) {
                        xAxis.add((String.valueOf(min_x + j * (max_x - min_x) / 10) + "-" + String.valueOf(min_x + (j + 1) * (max_x - min_x) / 10)));
                    }

                    int[] si = new int[10];
                    while (rs.next()) {
                        int c1 = rs.getInt(1);
                        int c2 = rs.getInt(2);
//                        System.out.printf("POST /data/onlinerec =========== c1=%d,c2=%d,min_x=%d,min_y=%d, max_x=%d,max_y=%d%n",c1,c2,min_x,min_y,max_x,max_y);
                        if ((c1 < max_x && c1 >= min_x) && (c2 < max_y && c2 >= min_y)) {
                            double delimetor = (double) (max_x - min_x) / (double) 10;
                            int index = (int) ((double) (c1 - min_x) / delimetor);
//                            System.out.printf("POST /data/onlinerec =========== delimetor=%f,result=%f%n",delimetor,result);
                            si[index]++;
                        }
                    }
                    for (int ii = 0; ii < 10; ii++) {
                        yAxis.add(si[ii]);
                    }
//                    if (true) {
//                        System.out.printf("POST /data/onlinerec =========== min=%d;max=%d%n", min, max);
//
//                        Collections.sort(si);
//                        int index = 0;
//                        while (index < si.size() && si.get(index) < min) ++index;
//                        for (int i = 0; i < 10; ++i) {
//                            xAxis.add((String.valueOf(min + i * (max - min) / 10) + "-" + String.valueOf(min + (i + 1) * (max - min) / 10)));
//                            int count = 0;
//                            while (index < si.size() && si.get(index) <= max && si.get(index) <= min + (i + 1) * (max - min) / 10) {
//                                ++index;
//                                ++count;
//                            }
//                            yAxis.add(count);
//                        }
//                    }
                } else if (type == Types.DOUBLE || type == Types.DECIMAL) {
                    double max_x = Double.valueOf((filter_x.get(1)).toString());
                    double min_x = Double.valueOf((filter_x.get(0)).toString());
                    int max_y = (int) filter_y.get(1);
                    int min_y = (int) filter_y.get(0);
                    System.out.printf("POST /data/onlinerec =========== type: DOUBLE%n");

                    for (int j = 0; j < 10; ++j) {
                        xAxis.add((String.valueOf(min_x + j * (max_x - min_x) / 10) + "-" + String.valueOf(min_x + (j + 1) * (max_x - min_x) / 10)));
                    }

                    int[] si = new int[10];
                    while (rs.next()) {
                        double c1 = rs.getDouble(1);
                        int c2 = rs.getInt(2);
//                        System.out.printf("POST /data/onlinerec =========== c1=%d,c2=%d,min_x=%d,min_y=%d, max_x=%d,max_y=%d%n",c1,c2,min_x,min_y,max_x,max_y);
                        if ((c1 < max_x && c1 >= min_x) && (c2 < max_y && c2 >= min_y)) {
                            double delimetor = (double) (max_x - min_x) / (double) 10;
                            int index = (int) ((double) (c1 - min_x) / delimetor);
//                            System.out.printf("POST /data/onlinerec =========== delimetor=%f,result=%f%n",delimetor,result);
                            si[index]++;
                        }
                    }
                    for (int ii = 0; ii < 10; ii++) {
                        yAxis.add(si[ii]);
                    }
                } else {
                    System.out.printf("POST /data/onlinerec =========== type: STRING%n");
                    int max_y = (int) filter_y.get(1);
                    int min_y = (int) filter_y.get(0);
                    Map<String, Integer> map = new HashMap<>();

                    for (int j = 0; j < filter_x.size(); j++) {
                        map.put((String) filter_x.get(j), 0);
                    }
                    try {
                        while (rs.next()) {
                            String c1 = rs.getString(1);
                            int c2 = rs.getInt(2);
                            if ((map.containsKey(c1)) && (c2 <= max_y && c2 >= min_y)) {
                                map.put(c1, map.get(c1) + 1);
                            }
                        }
                    } catch (SQLException e) {
                        e.printStackTrace();
                    }

                    List<Map.Entry<String, Integer>> list = new ArrayList<Map.Entry<String, Integer>>(map.entrySet());
//                            List<Map.Entry<String, Integer>> list = new ArrayList<Map.Entry<String, Integer>>(map.entrySet());
                    //非日期数据按条目数排序
                    Collections.sort(list, new Comparator<Map.Entry<String, Integer>>() {
                        @Override
                        public int compare(Map.Entry<String, Integer> o1, Map.Entry<String, Integer> o2) {
                            return o2.getValue().compareTo(o1.getValue());
                        }
                    });
                    Integer count = 0;
                    Integer sum = 0;
                    for (Map.Entry<String, Integer> entry : list) {
                        ++count;
                        if (count < 9) {
                            xAxis.add(entry.getKey());
                            yAxis.add(entry.getValue());
                        } else {
                            sum += entry.getValue();
                        }
                    }
                    if (count >= 10 && xAxis.size() <= 9) {
                        xAxis.add("Others");
                        yAxis.add(sum);
                    }
                }
            } catch (SQLException sqle) {
                sqle.printStackTrace();
            }

            Iterator<String> itrx = xAxis.listIterator();
            Iterator<Integer> itry = yAxis.listIterator();
            JSONObject dist = new JSONObject();
            dist.put("xcolumn", obj_x.getString("columnname"));
            dist.put("ycolumn", obj_y.getString("columnname"));
            Integer index = 0;
            JSONObject sec = new JSONObject();
            while (itrx.hasNext()) {
                ++index;
                JSONObject jobj = new JSONObject();
                jobj.put(itrx.next(), itry.next());
                sec.put(new String("sec" + index.toString()), jobj);
            }
            dist.put("data", sec);
            distArray.add(dist);
        }

//            Iterator<String> itrx = xAxis.listIterator();
//            Iterator<Integer> itry = yAxis.listIterator();
//            JSONArray jsarray = new JSONArray();
//            Integer index = 0;
//            while (itrx.hasNext()) {
//                ++index;
//                JSONObject jobj = new JSONObject();
//                jobj.put(itrx.next(),itry.next());
//                charOptions.put(new String("sec"+index.toString()),jobj);
//            }
        return new ResponseEntity<JSONArray>(distArray, HttpStatus.OK);
    }

    @CrossOrigin(origins = "*")
    @PostMapping("/data/deepeye")
    public ResponseEntity<?> GetDeepeyeRecommednation(@RequestBody String jsonData) {
        System.out.printf("POST /data/deepeye =========== jsonData=%s%n", jsonData);
        System.out.println("Saving Data As History =========");
//          SaveFilterAsHistory(JSONArray.fromObject(jsonData));
        JSONArray filter = JSONArray.fromObject(jsonData);
        JSONArray deepeye = new JSONArray();
        JSONArray resultArray = new JSONArray();
        int size = filter.size();
//
        String tablename = filter.getJSONObject(0).getString("tablename");
//
//          String columnName[] = new String[size];
//          Integer columnType[] = new Integer[size];
        String sql = "select * from " + tablename + " where 1=1 ";
        for (int i = 0; i < size; i++) {
            if (filter.getJSONObject(i).getInt("type") == 12) {
//                sql += "and " + filter.getJSONObject(i).getString("columnname") + ""
            } else if (filter.getJSONObject(i).getInt("type") == 93) {
                sql += "and " + filter.getJSONObject(i).getString("columnname") + " between '" + filter.getJSONObject(i).getJSONArray("data").get(0) + "' and '" + filter.getJSONObject(i).getJSONArray("data").get(1) + "'";
            } else {
                sql += "and " + filter.getJSONObject(i).getString("columnname") + " between '" + filter.getJSONObject(i).getJSONArray("data").get(0) + "' and '" + filter.getJSONObject(i).getJSONArray("data").get(1) + "'";
            }
        }
        sql = '"' + sql + '"';

        try {
            String args = new String("python partial_order_hive.py " + tablename + " " + sql);
            System.out.printf("args =========== %s%n", args);
            Process proc = Runtime.getRuntime().exec(args);
            BufferedReader in = new BufferedReader(new InputStreamReader(proc.getInputStream()));
            String line = null;
            int order = 0;
            while ((line = in.readLine()) != null && order < 4) {
                System.out.printf("======%s", line);
                JSONObject temp = JSONObject.fromObject(line);
                if (temp.getString("classify") == "scatter") {
                } else {
                    deepeye.add(temp);
                    order++;
                }
                System.out.println(deepeye);
            }
            in.close();
            proc.waitFor();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        JSONObject obj = new JSONObject();
        JSONObject result = new JSONObject();
        for (int i = 0; i < deepeye.size(); i++) {
            obj = deepeye.getJSONObject(i);
            result.put("xcolumn", obj.getString("x_name"));
            result.put("ycolumn", obj.getString("y_name"));
            result.put("cType", obj.getString("chart"));
            result.put("classify", obj.get("classify"));
            result.put("xdata", obj.get("x_data"));
            result.put("ydata", obj.get("y_data"));
            resultArray.add(result);
        }
        return new ResponseEntity<JSONArray>(resultArray, HttpStatus.OK);
    }

    @CrossOrigin(origins = "*")
    @PostMapping("/data/methodrec")
    public ResponseEntity<?> GetMethodRecommednation(@RequestBody String jsonData) {
        System.out.printf("POST /data/methodrec =========== jsonData=%s%n", jsonData);
        JSONArray resultArray = new JSONArray();

        JSONArray obj = JSONArray.fromObject(jsonData);
        String thisXcoulmn = obj.getJSONObject(1).getString("xcolumn");
        String stry = obj.getJSONObject(2).getString("ycolumn");
        String thisYcolumn = stry.substring(stry.indexOf("(") + 1, stry.indexOf(")"));
        String thisYtrans = stry.substring(0, stry.indexOf("("));
        System.out.printf("this Step2 Config : thisXcoulmn = %s , thisYcolumn = %s , thisYtrans = %s%n", thisXcoulmn, thisYcolumn, thisYtrans);

        HashMap<String, Double> methodRank = new HashMap<>();//用于存储各方法得分
        methodRank.put("exception detect", 0.0);
        methodRank.put("k-means", 0.0);
        methodRank.put("logistic regression", 0.0);
        HashMap<Integer, Double> similarity = new HashMap<>();//用于存储计算过的相似度，避免重复计算

        //计算相似度
        int objStep1Id = obj.getJSONObject(0).getInt("previousStepId");
        History h = historyRepository.findByLogId(objStep1Id);
        JSONArray thisFilter = new JSONArray();
        if (h != null) {
            String str = h.getRecord();
            thisFilter = JSONArray.fromObject(str);
            System.out.printf("this Step1 Filter : Filter = %s%n", thisFilter);
        }

        List<History> histories = historyRepository.findAllByStep(3);
        for (History history : histories) {
            JSONArray choosingMethod = JSONArray.fromObject(history.getRecord()).getJSONObject(1).getJSONArray("choosingMethod");

            Double similarityFromStep1 = 0.0;
            Double similarityFromStep2 = 0.0;
            int step1ID = -1;
            int step2ID = history.getPrevious();
//            String queryStep1ID = "select * from history where log_id = " + step2ID;
//            Statement st = conn.createStatement();
//            ResultSet qs1ID = st.executeQuery(queryStep1ID);
            History qs1ID = historyRepository.findByLogId(step2ID);
//            while (qs1ID.next()) {}
            if (qs1ID != null) {
                step1ID = qs1ID.getPrevious();
            }
            if (similarity.containsKey(step2ID)) {
                similarityFromStep2 = similarity.get(step2ID);
                similarityFromStep1 = similarity.get(step1ID);
            } else {
                //计算流程第二步之间的相似度
//                String queryStep2record = "select * from history where log_id = " + step2ID;
//                Statement stqs2r = conn.createStatement();
//                ResultSet rsStep2 = stqs2r.executeQuery(queryStep2record);
                History rsStep2 = historyRepository.findByLogId(step2ID);
//                while (rsStep2.next()) {
                if (rsStep2 != null) {
                    JSONArray tmp1 = JSONArray.fromObject(rsStep2.getRecord());
                    System.out.printf("history Step2 :  = %s%n", tmp1);
                    if (tmp1.size() > 1) {
                        String tmpXcoulmn = tmp1.getJSONObject(1).getString("xcolumn");
                        String tmpstry = tmp1.getJSONObject(2).getString("ycolumn");
                        String tmpYcolumn = tmpstry.substring(tmpstry.indexOf("(") + 1, tmpstry.indexOf(")"));
                        String tmpYtrans = tmpstry.substring(0, tmpstry.indexOf("("));
                        Double xNameIsEqual = (tmpXcoulmn == thisXcoulmn) ? 1 : 0.0;
                        Double yNameIsEqual = (tmpYcolumn == thisYcolumn) ? 1 : 0.0;
                        Double yTransIsEqual = (tmpYtrans == thisYtrans) ? 1 : 0.8;
                        similarityFromStep2 = 0.75 * xNameIsEqual + 0.25 * yNameIsEqual * yTransIsEqual;
                        similarity.put(step2ID, similarityFromStep2);
                    } else {
                        similarityFromStep2 = 0.00;
                        similarity.put(step2ID, similarityFromStep2);
                    }
                }
                //计算流程第一步之间的相似度
                if (similarity.containsKey(step1ID)) {
                    similarityFromStep1 = similarity.get(step1ID);
                } else {
//                    String queryStep1record = "select * from history where log_id = " + step1ID;
//                    Statement stqs1r = conn.createStatement();
//                    ResultSet rsStep1 = stqs1r.executeQuery(queryStep1record);
                    History rsStep1 = historyRepository.findByLogId(step1ID);
//                    while (rsStep1.next()) {
                    if (rsStep1 != null) {
                        JSONArray tmp2 = JSONArray.fromObject(rsStep1.getRecord());
                        System.out.printf("history Step1 :  = %s%n", tmp2);
                        if (thisFilter.size() != 0) {
                            for (int i = 0; i < tmp2.size(); i++) {
                                if (tmp2.getJSONObject(i).getInt("type") != 12) {
                                    String thisdata = thisFilter.getJSONObject(i).getString("data");
                                    String thismin = thisdata.substring(thisdata.indexOf("[") + 1, thisdata.indexOf(","));
                                    String thismax = thisdata.substring(thisdata.indexOf(",") + 1, thisdata.indexOf("]"));

                                    String tmpdata = tmp2.getJSONObject(i).getString("data");
                                    String tmpmin = tmpdata.substring(tmpdata.indexOf("[") + 1, tmpdata.indexOf(","));
                                    String tmpmax = tmpdata.substring(tmpdata.indexOf(",") + 1, tmpdata.indexOf("]"));

                                    switch (tmp2.getJSONObject(i).getInt("type")) {
                                        case 93:
                                            similarityFromStep1 += 0.0;
                                            break;
                                        case 4:
                                            int thmin = Integer.parseInt(thismin);
                                            int thmax = Integer.parseInt(thismax);
                                            int tmmin = Integer.parseInt(tmpmin);
                                            int tmmax = Integer.parseInt(tmpmax);
                                            double ratio;
                                            if (thmin == thmax || tmmin == tmmax) {
                                                ratio = (thmax - thmin) + 1 / (tmmax - tmmin) + 1;
                                            } else {
                                                ratio = (thmax - thmin) / (tmmax - tmmin);
                                            }
                                            if (ratio > 1) ratio = 1 / ratio;

                                            double overlap = 0.0;
                                            if (!(thmax < tmmin || tmmax < thmin)) {
                                                double ol = tmmax < thmax ? tmmax - thmin : thmax - tmmin;
                                                overlap = (ol / (tmmax - tmmin) + ol / (thmax - thmin)) / 2;
                                                //System.out.printf("%.5f , %,5f%n", ratio, overlap);
                                            }
                                            //System.out.printf("again , and this is int . %.5f , %,5f%n", ratio, overlap);
                                            similarityFromStep1 += (0.7 * ratio + 0.3 * overlap) / 6;
                                            //System.out.printf("the similarity between [%d,%d] and [%d,%d] is %.5f%n", thmin, thmax, tmmin, tmmax, (0.7 * ratio + 0.3 * overlap));
                                            break;
                                        case 8:
                                            Double thminD = Double.parseDouble(thismin);
                                            Double thmaxD = Double.parseDouble(thismax);
                                            Double tmminD = Double.parseDouble(tmpmin);
                                            Double tmmaxD = Double.parseDouble(tmpmax);
                                            double ratioD;
                                            if (thminD == thmaxD || tmminD == tmmaxD) {
                                                ratioD = (thmaxD - thminD) + 1 / (tmmaxD - tmminD) + 1;
                                            } else {
                                                ratioD = (thmaxD - thminD) / (tmmaxD - tmminD);
                                            }
                                            if (ratioD > 1) ratioD = 1 / ratioD;

                                            double overlapD = 0.0;
                                            if (!(thmaxD < tmminD || tmmaxD < thminD)) {
                                                double olD = tmmaxD < thmaxD ? tmmaxD - thminD : thmaxD - tmminD;
                                                overlapD = (olD / (tmmaxD - tmminD) + olD / (thmaxD - thminD)) / 2;
                                                //System.out.printf("%.5f , %,5f%n", ratioD, overlapD);
                                            }
                                            //System.out.printf("again, and this is double. %.5f , %,5f%n", ratioD, overlapD);
                                            similarityFromStep1 += (0.7 * ratioD + 0.3 * overlapD) / 6;
                                            //System.out.printf("the similarity between [%.2f,%.2f] and [%.2f,%.2f] is %.5f%n", thminD, thmaxD, tmminD, tmmaxD, 0.7 * ratioD + 0.3 * overlapD);
                                            break;
                                    }
                                }
                            }
                        }
                    }
                    System.out.print(step1ID);
                    System.out.println(similarityFromStep1);

                }
            }
            System.out.printf("similarity in step1 : %s and step2 : %s is %.5f , %.5f%n", step1ID, step2ID, similarityFromStep1, similarityFromStep2);
            Double thisSimilarity = 0.0;
            if (similarityFromStep1 != null && similarityFromStep2 != null) {
                thisSimilarity = 0.6 * similarityFromStep2 + 0.4 * similarityFromStep1;//流程相似度
            }
            for (int i = 0; i < choosingMethod.size(); i++) {
                String methodstr = (String) choosingMethod.get(i);
                methodRank.put(methodstr, methodRank.get(methodstr) + thisSimilarity);
            }
        }
//        } catch (Exception e)
//        {e.printStackTrace();}
        //对hashmap排序，降序输出方法和得分
        List<Map.Entry<String, Double>> list = new ArrayList<>(methodRank.entrySet());
        Collections.sort(list, new Comparator<Map.Entry<String, Double>>() {
            // 根据value排序
            public int compare(Map.Entry<String, Double> o1, Map.Entry<String, Double> o2) {
                double result = o2.getValue() - o1.getValue();
                if (result > 0)
                    return 1;
                else if (result == 0)
                    return 0;
                else
                    return -1;
            }
        });
//        for (Map.Entry<String, Double> set : list) {
//            System.out.println(set.getKey() + "   " + set.getValue());
//            JSONObject tmp3 = new JSONObject();
//            tmp3.put(set.getKey(), set.getValue());
//            resultArray.add(tmp3);
//        }
        //根据选择的列，返回不同方法
        int xType=0, yType = 0;
        switch (thisXcoulmn) {
            case "solddate":
                xType = 1;//时序
                break;
            case "age":
            case "country":
            case "nationality":
                xType = 2;//属性
                break;
            case "customer":
            case "itemname":
                xType = 3;//主体
                break;

        }
        if (thisYcolumn.contains("quantity")|| thisYcolumn.contains("price") || thisYcolumn.contains("discount")) {
            yType = 1;
        }
        List<String> classification = new ArrayList<String>(){{add("分类");add("对买家客户做分类");add("订制个性化销售策略");add("优化在售商品结构");}};
        List<String> outlierdetection = new ArrayList<String>(){{add("诊断");add("诊断店铺经营状况");add("找出异常销售数据");add("发现店铺隐藏问题");}};
        List<String> regression = new ArrayList<String>(){{add("预测");add("按需调整进货数量");add("控制商品库存计划");add("设计促销活动保证销量");}};
        List<String> clustering = new ArrayList<String>(){{add("探索");add("了解买家客户画像");add("找准店铺经营定位");add("识别店铺流量来源");}};
        System.out.printf("xcolumn==%s, ycolumn===%s, xtype===%d, ytype===%d",thisXcoulmn,thisYcolumn,xType,yType);
        if(yType==1&&xType==1){
            resultArray.add(regression);resultArray.add(outlierdetection);resultArray.add(classification);
        }else if(xType==2){
            resultArray.add(classification);resultArray.add(clustering);resultArray.add(outlierdetection);
        }else if(xType==3){
            resultArray.add(clustering);resultArray.add(classification);resultArray.add(outlierdetection);
        }else{
            resultArray.add(outlierdetection);resultArray.add(regression);resultArray.add(classification);
        }

        return new ResponseEntity<>(resultArray, HttpStatus.OK);
    }

    //The API is obsoleted.
    @CrossOrigin(origins = "*")
    @PostMapping("/data/offlinedist")
    public ResponseEntity<?> GetOfflineDistribution(@RequestBody String jsonData) {
        JSONObject obj = JSONObject.fromObject(jsonData);
        System.out.printf("POST /data/distribution =========== tablename=%s%n", obj.getString("tablename"));
        return new ResponseEntity<>(dataDistributionRepository.findBySource(obj.getString("tablename")), HttpStatus.OK);
    }

    @CrossOrigin(origins = "*")
    @PostMapping("/data/range")
    public ResponseEntity<?> RecommendColumnsAndGetRange(@RequestBody String jsonData) {
        System.out.printf("POST /data/range =========== parameter=%s%n", jsonData);

        JSONObject obj = JSONObject.fromObject(jsonData);
        System.out.printf("POST /data/range =========== tablename=%s%n", obj.getString("tablename"));
        /*
        *********************************************************************
            The algorithm for column recommendation
        *********************************************************************
        */
        ArrayList<String> columns = new ArrayList<String>();
//        String filter = null;
        String lastline = null;
        try {
            String[] args = new String[]{"python", "./single-row-chart/__init__.py", obj.getString("tablename")};
//            String[] args = new String[]{"python", "/Users/frankqian/Frank/Study/Master/智能推荐/Smart_interaction_Project/smartinteract/single-row-chart/__init__.py"};

            Process proc = Runtime.getRuntime().exec(args);
            System.out.println("__init__.py succeeded");

            BufferedReader in = new BufferedReader(new InputStreamReader(proc.getInputStream()));
            String line = null;
            Integer rec_num = 0;
            while ((line = in.readLine()) != null) {
                System.out.printf("Recommend ====== %s%n", line);
                if (rec_num < 2)
                    columns.add(line.substring(line.indexOf(')') + 2, line.indexOf(' ', line.indexOf(')') + 2)));
                ++rec_num;
                lastline = line;
            }
            if (columns.isEmpty()) {
                System.out.println("Recommend =========== No recommend result!");
                return new ResponseEntity<>(HttpStatus.NO_CONTENT);
            }
//            filter = lastline;
            else {
                System.out.println("Recommend 1 ====== " + columns.get(0));
                System.out.println("Recommend 2 ====== " + columns.get(1));
                System.out.println("Recommend filter ====== " + lastline);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }

        /*
         *********************************************************************
         */
//        lastline = "[['price', [0.0, 54.582]], ['solddate', ['2005-11-12', '2005-12-13']]]";
        int pos = -1;
        pos = lastline.indexOf(']', lastline.indexOf(']') + 1) + 1;
        String[] filter = {"", ""};
//        String[] filter = {"['price', 0.0, 54.582]","['solddate', datetime.datetime(2005, 11, 12, 0, 0), datetime.datetime(2005, 12, 13, 0, 0)]"};
        filter[0] = lastline.substring(lastline.indexOf(',') + 3, pos - 2);
        filter[1] = lastline.substring(lastline.indexOf('[', pos + 3) + 1, lastline.length() - 3);

        System.out.println(filter[0]);
        System.out.println(filter[1]);
        ArrayList<String> columnNames = new ArrayList<String>();
//        JSONArray columns = obj.getJSONArray("columns");
        int num = columns.size();
//        int num=2;
        //Assemble the sql query
        String sql = "SELECT ";
        for (int i = 0; i < num; i++) {
            String s = columns.get(i);
            columnNames.add(s);
            sql += s;
            if (i != num - 1) sql += ",";
        }
        sql += " FROM " + obj.getString("tablename");
        System.out.printf("POST /data/range =========== sql=%s%n", sql);
        try {
            PreparedStatement ps = HiveConfig.prepare(conn, sql);
            ResultSet rs = ps.executeQuery();

            ArrayList<Integer> columnTypes = new ArrayList<>();
            for (int i = 1; i <= num; i++) {
                columnTypes.add(rs.getMetaData().getColumnType(i));
            }

            HashSet<?>[] data = new HashSet<?>[num];
            for (int i = 0; i < num; i++) {
                if (columnTypes.get(i) == Types.BIGINT || columnTypes.get(i) == Types.INTEGER) {
                    data[i] = new HashSet<Integer>();
                } else if (columnTypes.get(i) == Types.DOUBLE || columnTypes.get(i) == Types.REAL || columnTypes.get(i) == Types.FLOAT) {
                    data[i] = new HashSet<Double>();
                } else if (columnTypes.get(i) == Types.TIMESTAMP) {
                    data[i] = new HashSet<Timestamp>();
                } else {
                    data[i] = new HashSet<String>();
                }
            }
            while (rs.next()) {
//                System.out.println("alive");
                for (int i = 0; i < num; i++) {
                    if (columnTypes.get(i) == Types.BIGINT || columnTypes.get(i) == Types.INTEGER) {
                        ((HashSet<Integer>) data[i]).add(rs.getInt(i + 1));
                    } else if (columnTypes.get(i) == Types.DOUBLE || columnTypes.get(i) == Types.REAL || columnTypes.get(i) == Types.FLOAT) {
                        ((HashSet<Double>) data[i]).add(rs.getDouble(i + 1));
                    } else if (columnTypes.get(i) == Types.TIMESTAMP) {
//                        System.out.println(rs.getTimestamp(i+1));
                        ((HashSet<Timestamp>) data[i]).add(rs.getTimestamp(i + 1));
                    } else {
                        ((HashSet<String>) data[i]).add(rs.getString(i + 1));
                    }
                }
            }
            JSONObject returnData = new JSONObject();

            JSONArray recommData = new JSONArray();
            for (int i = 0; i < num; i++) {
                JSONObject column = new JSONObject();
                column.put("columnname", columnNames.get(i));
                column.put("tablename", obj.getString("tablename"));
                int type = columnTypes.get(i);
                column.put("type", type);
                if (type == Types.BIGINT || type == Types.INTEGER) {
                    int min = Collections.min((HashSet<Integer>) data[i]);
                    int max = Collections.max((HashSet<Integer>) data[i]);
                    int minn = Integer.parseInt(filter[i].split(",")[0]);
                    int maxx = Integer.parseInt(filter[i].split(",")[1]);
                    int[] range = {min, max};
                    column.put("range", range);

                    int[] recomm = {minn, maxx};
                    column.put("data", recomm);
                } else if (type == Types.DOUBLE || type == Types.REAL || type == Types.FLOAT) {
                    double min = Collections.min((HashSet<Double>) data[i]);
                    double max = Collections.max((HashSet<Double>) data[i]);
                    double minn = Double.parseDouble(filter[i].split(",")[0]);
                    double maxx = Double.parseDouble(filter[i].split(",")[1]);
                    double[] range = {min, max};
                    column.put("range", range);

                    double[] recomm = {minn, maxx};
                    column.put("data", recomm);
                } else if (type == Types.TIMESTAMP) {
                    Timestamp min = Collections.min((HashSet<Timestamp>) data[i]);
                    Timestamp max = Collections.max((HashSet<Timestamp>) data[i]);
                    String minn = filter[i].split(", |,")[0].substring(1, filter[i].split(",")[0].length() - 1) + " 00:00:00.0";
                    String maxx = filter[i].split(", |,")[1].substring(1, filter[i].split(",")[0].length() - 1) + " 00:00:00.0";
                    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.S");
                    System.out.println(minn);
                    System.out.println(maxx);
                    String sf = sdf.format(min);
                    String sf1 = sdf.format(max);
                    String[] range = {sf, sf1};
                    column.put("range", range);


                    String[] recomm = {minn, maxx};
                    column.put("data", recomm);
//                    System.out.println(min);
//                    System.out.println(max);
//                    System.out.println(recomm[0]);
                } else {
                    Iterator itr = data[i].iterator();
                    int index = 0;
                    Map<String, Integer> map = new HashMap<>();
                    ArrayList<String> arr = new ArrayList<>();
                    while (itr.hasNext() && index < 2) {
                        String str = (String) itr.next();
                        if (map.containsKey(str)) {
                            map.put(str, map.get(str) + 1);
                        } else {
                            map.put(str, 1);
                        }
                    }
                    List<Map.Entry<String, Integer>> list = new ArrayList<Map.Entry<String, Integer>>(map.entrySet());
                    Collections.sort(list, new Comparator<Map.Entry<String, Integer>>() {
                        @Override
                        public int compare(Map.Entry<String, Integer> o1, Map.Entry<String, Integer> o2) {
                            return o1.getKey().compareTo(o2.getKey());
                        }
                    });
                    Integer count = 0;
                    Integer sum = 0;
                    for (Map.Entry<String, Integer> entry : list) {
                        ++count;
                        if (count <= 9) {
                            arr.add(entry.getKey());
//                            xAxis[index].add(entry.getKey());
//                            yAxis[index].add(entry.getValue());
                        } else if (count >= 10) {
                            break;
//                            sum+=entry.getValue();
                        }
                    }

                    if (count >= 10 && arr.size() <= 9) {
                        arr.add("Others");
//                        yAxis[index].add(sum);
                    }

                    column.put("range", data[i]);
                    column.put("data", arr);
                }
                recommData.add(column);

            }
            returnData.put("recommend", recommData);

            //Calculate all the column distributions under the filtering condition.
            returnData.put("distribution", GetAllDataDistributions(recommData));

            return new ResponseEntity<JSONObject>(returnData, HttpStatus.OK);
        } catch (SQLException e1) {
            e1.printStackTrace();
        }

        return new ResponseEntity<>(HttpStatus.NO_CONTENT);
    }

    @CrossOrigin(origins = "*")
    @PostMapping("/hive/updatedata")
    public DeferredResult<ResponseEntity<?>> UpdateTableData(@RequestBody String jsonData) throws
            ClassNotFoundException, SQLException {
        System.out.printf("POST /hive/updatedata =========== parameter=%s%n", jsonData);
        JSONObject obj = JSONObject.fromObject(jsonData);
        DeferredResult<ResponseEntity<?>> result = new DeferredResult<>();

//        String sql = "SELECT distinct itemname FROM " + obj.getString("tablename");
//        HashSet<String> hs = new HashSet<>();
//        try {
//            PreparedStatement ps = HiveConfig.prepare(conn, sql);
//            ResultSet rs = ps.executeQuery();
//            while (rs.next()) {
//                hs.add(rs.getString(1));
//            }
//        } catch (SQLException e) {
//            e.printStackTrace();
//        }
//
//        DeferredResult<ResponseEntity<?>> result = new DeferredResult<>();
//
//        new Thread(() -> {
//            result.setResult(ResponseEntity.ok(now()));
//
//            try {
//                Statement s = conn.createStatement();
//                Iterator iter = hs.iterator();
//                while (iter.hasNext()) {
//                    String item = (String) iter.next();
//                    System.out.printf("POST /hive/updatedata =========== item=%s%n", item);
//                    String item1 = item.replaceFirst("^A*", "");
//                    System.out.printf("POST /hive/updatedata =========== item1=%s%n", item1);
//                    if (!item1.equals(item)) {
//                        String sql1 = "UPDATE " + obj.getString("tablename") + " SET itemname='" + item1 + "' WHERE itemname='" + item + "'";
//                        System.out.printf("POST /hive/updatedata =========== sql1=%s%n", sql1);
//                        s.executeUpdate(sql1);
//                    }
//                }
//            } catch (SQLException e1) {
//                e1.printStackTrace();
//            }
//        }, "MyThread-").start();
//        return result;

        String sql = "UPDATE " + obj.getString("tablename") + " SET price=" + "0" + " WHERE solddate>='2005-02-28' and solddate<='2005-04-02'";
        System.out.println(sql);
// HashSet<String> hs = new HashSet<>();
        try {
            PreparedStatement ps = HiveConfig.prepare(conn, sql);
            ps.execute(sql);
        } catch (SQLException e) {
            e.printStackTrace();
        }

        return result;
    }

    @CrossOrigin(origins = "*")
    @PostMapping("/hive/updatedesc")
    public DeferredResult<ResponseEntity<?>> UpdateItemDesc(@RequestBody String jsonData) throws
            ClassNotFoundException, SQLException {
        System.out.printf("POST /hive/updatedesc =========== parameter=%s%n", jsonData);
        JSONObject obj = JSONObject.fromObject(jsonData);

        String sql = "SELECT itemdesc FROM " + obj.getString("tablename");
        HashSet<String> hs = new HashSet<>();
        try {
            PreparedStatement ps = HiveConfig.prepare(conn, sql);
            ResultSet rs = ps.executeQuery();
            while (rs.next()) {
                hs.add(rs.getString(1));
            }
        } catch (SQLException e) {
            e.printStackTrace();
        }
        System.out.printf("POST /hive/updatedesc =========== size=%s%n", hs.size());
        DeferredResult<ResponseEntity<?>> result = new DeferredResult<>();

        new Thread(() -> {
            result.setResult(ResponseEntity.ok(now()));
            try {
                PreparedStatement s = conn.prepareStatement(" UPDATE web_sales_home1 SET itemdesc=? WHERE itemdesc=?");
                Iterator iter = hs.iterator();
                while (iter.hasNext()) {
                    String itemdesc = (String) iter.next();
                    if (itemdesc.indexOf('\'') != -1) {
                        String sdesc = itemdesc.replaceAll("'", "");
                        System.out.printf("POST /hive/updatedesc =========== itemdesc=%s%n", itemdesc);
                        s.setString(1, sdesc);
                        s.setString(2, itemdesc);
                        s.executeUpdate();
                    }

                }
            } catch (SQLException e1) {
                e1.printStackTrace();
            }
        }, "MyThread-").start();
        return result;
    }

    @CrossOrigin(origins = "*")
    @PostMapping("/data/exceptiondetect")
    public ResponseEntity<?> DetectExceptionSectionInSequece(@RequestBody String jsonData) {
        JSONArray jarray = JSONArray.fromObject(jsonData);
        JSONArray answer = new JSONArray();
//       System.out.printf("POST /data/distribution =========== tablename=%s%n", obj.getString("tablename"));
        System.out.printf("==================/data/exceptiondetect==============%n");
//          for (int i=0;i<jarray.size();++i) {
//            JSONObject obj = jarray.getJSONObject(i);
        JSONObject obj = jarray.getJSONObject(0);
        JSONObject data = obj.getJSONObject("data");
        /*
            *********************************************************************
            The algorithm for exception detection in sequence data
            *********************************************************************
            */
        IForest iForest = new IForest();
        double[][] samples = new double[data.size()][2];
        Integer maxsec = 0;
        for (; data.containsKey("sec" + Integer.toString(maxsec + 1)); ++maxsec) {
            JSONObject obj1 = data.getJSONObject("sec" + Integer.toString(maxsec + 1));
            Collection collection = obj1.values();
            samples[maxsec][0] = Double.valueOf(collection.iterator().next().toString());
            samples[maxsec][1] = samples[maxsec][0];
//            System.out.println(samples[maxsec] [0]);
        }
        try {
            int[] labels = iForest.train(samples, 100);
            int n = 0;
            for (int label : labels) {
                System.out.print(label + " ");
                if ((n + 1) % 10 == 0)
                    System.out.println();
                n++;
            }
            Integer sum_of_exception_point = labels.length;
            JSONObject jobj = new JSONObject();
            Integer[] p = new Integer[maxsec];
            for (Integer j = 0; j < maxsec; ++j) p[j] = 0;
            for (Integer j = 0; j < sum_of_exception_point; ++j) {
                if (labels[j] == 0) {
                    jobj.put("Exp" + Integer.toString(j + 1), "sec" + Integer.toString(j + 1));
                }
            }
            obj.put("Exception", jobj);
            answer.add(obj);

        } catch (Exception e) {
            e.printStackTrace();
        }

        return new ResponseEntity<JSONArray>(answer, HttpStatus.OK);
    }


    @CrossOrigin(origins = "*")
    @PostMapping("/data/datalineage")
    public ResponseEntity<?> DataLineage(@Valid @RequestBody JSONObject json) {
        Integer userid = json.getInt("user_id");
        Integer curLogId = json.getInt("log_id");
        System.out.printf("==================/data/datalineage==============%n");

        Statement stmt1 = null;
        String url = "jdbc:mysql://10.141.223.30:3306/userdb?characterEncoding=utf8&useSSL=true";
        String user = "root";
        String pass = "root";
        String sql = "";
        String sql1 = "";
        String sqlhis = "";
        JSONObject answer = new JSONObject();
        try {
            Class.forName("com.mysql.jdbc.Driver");
            Connection conn = DriverManager.getConnection(url, user, pass);
            Statement stmt = conn.createStatement(ResultSet.TYPE_SCROLL_SENSITIVE, ResultSet.CONCUR_UPDATABLE);
            //sql = "SELECT  * FROM history where user_id=" + userid + " order by dataset_id ";
            sql = "SELECT  * FROM (select * from userdb.history order by log_id desc limit 10) as a\n" +
                    "where user_id= "+userid + " order by log_id; ";
            ResultSet rs = stmt.executeQuery(sql);
            rs.next();
            Integer dataset_id = rs.getInt(3);
//            Integer  stepnum = 0, nextstep, nextnextstep, previous;
//            String str_record;
            //JSON group by dataset
//            Integer dataset_id = -1;
            JSONArray array = new JSONArray();
            answer.put("name", "原始数据");
            answer.put("deal", 2);
//
//                    }
//                } else {
//                    if (dataset_id == -1) {
//                        dataset_id = curid;
//                    } else {
//                        sql1 = "SELECT tablename FROM sourceindex WHERE dataset_id=" + dataset_id;
//                        stmt1 = conn.createStatement();
//                        ResultSet tablers = stmt1.executeQuery(sql1);
//                        tablers.next();
//                        String tablename = tablers.getString("tablename");
//                        JSONObject jsonObject = new JSONObject();
//                        jsonObject.put("name", tablename);
//                        if (!tempanswer.isEmpty()) {
//                            jsonObject.put("children", tempanswer);
//                            array.add(jsonObject);
//                            tempanswer = new JSONArray();
//                        }
//                        dataset_id = curid;
//                    }
//                    historyRsultSet.previous();
//                }
//            }

//            //last dataset tempanswer to answer
//            if (!tempanswer.isEmpty()) {
//                sql1 = "SELECT tablename FROM sourceindex WHERE dataset_id=" + dataset_id;
//                stmt1 = conn.createStatement();
//                ResultSet tablers = stmt1.executeQuery(sql1);
//                tablers.next();
//                String tablename = tablers.getString("tablename");
//                JSONObject jsonObject = new JSONObject();
//                jsonObject.put("name", tablename);
//                if (!tempanswer.isEmpty()) {
//                    jsonObject.put("children", tempanswer);
//                    array.add(jsonObject);
//                    tempanswer = new JSONArray();
//                }
//            }
            sql = "SELECT tablename,row_count FROM sourceindex WHERE dataset_id=" + dataset_id;
            stmt1 = conn.createStatement();
            ResultSet tablers = stmt1.executeQuery(sql);
            tablers.next();
            //String tablename = tablers.getString("tablename");
            String tablename = "商家数据";
            Integer rowCount = tablers.getInt("row_count");
            JSONObject jsonObject = new JSONObject();
            jsonObject.put("name", tablename);
            jsonObject.put("rowcount",rowCount);
            jsonObject.put("deal", 2);
            JSONArray tempArray = new JSONArray();
            rs.previous();
            while (rs.next()) {
                Integer log_id = rs.getInt(1);
                Integer user_id = rs.getInt(2);
                Integer step = rs.getInt(4);
                String str_record = rs.getString(5);
                JSONArray record = JSONArray.fromObject(str_record);
                Integer previous = rs.getInt(6);
                while (previous!=-1){
                    rs.next();
                }
                Timestamp time = rs.getTimestamp(7);
                rowCount=rs.getInt(8);
                //System.out.println("logid:" + rs.getInt(1));
                JSONObject jo1 = new JSONObject();
                JSONArray ja1 = new JSONArray();
                JSONObject jo2 = new JSONObject();
                JSONArray ja2 = new JSONArray();
                JSONObject jo3 = new JSONObject();
                JSONArray ja3 = new JSONArray();
                String filter = "子数据集筛选:";
                int f=0;
                for (int i = 0; i < record.size(); i++) {
                    if (!record.getJSONObject(i).getString("data").equals("[]")) {
                        if (!record.getJSONObject(i).getString("columnname").contains("1")){
                            String colname = record.getJSONObject(i).getString("columnname");
                            filter += colname + ",";
                            f++;
                        }
                    }
                }
                filter+=" "+f+"列";
                jo1.put("name", filter);
                jo1.put("rowcount",rowCount);
                if (log_id.equals(curLogId)){
                    jo1.put("deal", 2);
                }
                while (rs.next()) {
                    log_id = rs.getInt(1);
                    //System.out.println("logid:"+log_id);
                    step = rs.getInt(4);
                    str_record = rs.getString(5);
                    record = JSONArray.fromObject(str_record);
                    previous = rs.getInt(6);
                    rowCount=rs.getInt(8);
                    if (previous == -1) {
                        rs.previous();
                        break;
                    } else if (step == 2) {
                        String str = "可视化选取:" + "X轴="+record.getJSONObject(1).get("xcolumn") + ",Y轴=" + record.getJSONObject(2).get("ycolumn");
                        jo2.put("name", str);
                        jo2.put("rowcount",rowCount);
                        if (log_id.equals(curLogId)){
                            jo1.put("deal", 2);
                            jo2.put("deal", 2);
                        }
                        if (jo2 != null) {
                            ja2.add(jo2);
                        } else {
                            System.out.println("jo2 NULL");
                        }
                    } else {
                        filter = rs.getString(5);
                        filter = filter.replace("\":\"", "->");
                        jo3.put("name", filter);
                        jo3.put("rowcount",rowCount);
                        if (log_id.equals(curLogId)){
                            jo1.put("deal", 2);
                            jo2.put("deal", 2);
                            jo3.put("deal", 2);
                        }
                        ja3.add(jo3);
                    }
                }
                if (ja2.size() >= 1) {
                    jo1.put("children", ja2);
                }
                if (jo1.size() >= 1) {
                    tempArray.add(jo1);
                } else {
                    System.out.println("Error jo1 NULL");
                }

            }
            jsonObject.put("children", tempArray);
            array.add(jsonObject);
            answer.put("children", array);

//            //getrecnthistory
//            Connection connhis = DriverManager.getConnection(url, user, pass);
//            Statement stmthis = conn.createStatement(ResultSet.TYPE_SCROLL_SENSITIVE, ResultSet.CONCUR_UPDATABLE);
//            JSONObject result = new JSONObject();
//            sqlhis = "SELECT * FROM history WHERE user_id = " + userid.toString() + " ORDER BY log_id DESC LIMIT 1";
//            ResultSet rs1 = stmthis.executeQuery(sqlhis);
//            boolean ispreroot = false;
//            ArrayList<String> names = new ArrayList<>();
//            Integer rowcount=-1;
//            while (!ispreroot) {
//                rs1.next();
//                dataset_id = rs1.getInt(3);
//                Integer step = rs1.getInt(4);
//                String str_record = rs1.getString(5);
//                JSONArray record = JSONArray.fromObject(str_record);
//                Integer previous = rs1.getInt(6);
//                rowcount = rs1.getInt(8);
//                sql = "SELECT tablename FROM sourceindex WHERE dataset_id=" + dataset_id;
//                ResultSet rs2 = stmt.executeQuery(sql);
//                rs2.next();
//                tablename = rs2.getString("tablename");
//                result.put("name", tablename);
//                String name = "";
//                if (step == 1) {
//                    for (int i = 0; i < record.size(); i++) {
//                        if (!record.getJSONObject(i).getJSONArray("data").isEmpty())
//                            name = name + record.getJSONObject(i).getString("columnname") + ":" + record.getJSONObject(i).getJSONArray("data").toString() + " ";
//                    }
//                } else if (step == 2) {
//                    name = name + "xcolumn,ycolumn:" + record.getJSONObject(1).getString("xcolumn") + "," + record.getJSONObject(2).getString("ycolumn");
//                } else {
//                    name = name + record.toString().replace("\":\"", "->");
//                }
//                names.add(name);
//
//                if (previous == -1) ispreroot = true;
//                else {
//                    sqlhis = "SELECT * FROM history WHERE user_id = " + userid.toString() + " AND log_id = "+ previous.toString();
//                    rs1 = stmthis.executeQuery(sqlhis);
//                }
//            }
//            //System.out.println(names.size());
//            JSONArray jsonArray = new JSONArray();
//            for (int i = 0; i < names.size(); i++) {
//                jsonObject = new JSONObject();
//                jsonObject.put("name", names.get(i));
//                jsonObject.put("deal", 2);
//                jsonObject.put("rowcount", rowcount);
//                if (i != 0) {
//                    jsonObject.put("children", jsonArray);
//                }
//                jsonArray = new JSONArray();
//                jsonArray.add(jsonObject);
//            }
//            result.put("children", jsonArray);
////
//            for (int i = 0; i < answer.getJSONArray("children").size(); i++) {
//                if (answer.getJSONArray("children").getJSONObject(i).get("name").equals(result.get("name"))) {
//                    answer.getJSONArray("children").getJSONObject(i).put("deal", 2);
//                    answer.getJSONArray("children").getJSONObject(i)
//                            .getJSONArray("children").add(jsonArray.getJSONObject(0));
//                }
//            }
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        } catch (SQLException e) {
            e.printStackTrace();
        }
        return new ResponseEntity<>(answer, HttpStatus.OK);
    }

}
